
#include	<string.h>
#include    <unistd.h>
#include	<errno.h>
#include	<time.h>
#include	<ctype.h>
#include	<stdexcept>

#include	"fifo_storage.hpp"

#include	"wbsinker_error.hpp"



#define	MAX_SLICE_SIZE		(256 * 1048576)	// 256MB

#define	DEFAULT_SWITCH_TIMEOUT		300

#define	DEFAULT_SWITCH_COUNT_LIMIT	1000


FifoStorage::FifoStorage ( const char* pszConfigFile )
	:	m_iSwitchCountLimit(DEFAULT_SWITCH_COUNT_LIMIT), 
		m_iSwitchTimeout(DEFAULT_SWITCH_TIMEOUT),
		m_iSwitchSliceSize(MAX_SLICE_SIZE),
		m_iOutCacheSize(FS_CACHE_SIZE_DEFAULT), m_iOutCacheSizeFree(FS_CACHE_SIZE_DEFAULT),
		m_iOutCacheSizeUsed(0), m_pszOutCache(NULL),
		m_bCIDPrefix(true), m_bDateSubdir(true), m_bGZipFile(false),
		m_pszSinkRoot( NULL ),
		m_pszConfigFile(NULL)
{
	m_iSwitchSliceSize = MAX_SLICE_SIZE;
	
	loadConfig( pszConfigFile );

	m_strFifoList = std::string(m_pszSinkRoot) + "/" + "fifo.list";

	m_strDataDir = std::string(m_pszSinkRoot) + "/fifodata";
	
	m_iFlushBufSize = FS_FLUSH_BUF_SIZE_DEFAULT;
	m_pszFlushBuf = (char *)malloc( FS_FLUSH_BUF_SIZE_DEFAULT * sizeof(char) );
	if ( NULL == m_pszFlushBuf ) {
		throw std::runtime_error( "Fail to malloc flush-buffer." );
	}
	
	int	_iLen = sprintf( m_szSwitchSuffix, "fifo" );
	m_szSwitchSuffix[ _iLen ] = '\0';
	
	mkpdir ( m_strDataDir.c_str() );
}

FifoStorage::~FifoStorage ()
{
	this->close ();
	
	if( NULL != m_pszFlushBuf ) {
		free( m_pszFlushBuf );
		m_pszFlushBuf = NULL;
	}	

	if (NULL != m_pszSinkRoot) {
		free( m_pszSinkRoot );
		m_pszSinkRoot = NULL;
	}
}

int
FifoStorage::open( void )
{
/*
#ifdef	__linux__
	sprintf ( szBuf, "%s/%lu", m_strDataDir.c_str(), time(NULL) );
#else
#if	__GNUC__ < 3
	sprintf ( szBuf, "%s/%ld", m_strDataDir.c_str(), time(NULL) );
#else
	sprintf ( szBuf, "%s/%d", m_strDataDir.c_str(), time(NULL) );
#endif
#endif
*/
	return	0;
}

int
FifoStorage::close ( void )
{
	//DEBUG ( CAT_SINKER, "ENTER FifoStorage::close()." );
	
	this->flushCache ();
	
	//DEBUG ( CAT_SINKER, "Cache-flushed  FifoStorage::close()." );

	map_storage_cell_t::iterator	it = m_mapStorageCell.begin();
	while( m_mapStorageCell.end() != it ) {
		int		iType = it->first;
		
		this->close( iType );
		
		++it;
	}
	
	//DEBUG ( CAT_SINKER, "EXIT  FifoStorage::close()." );
	return	0;
}

int
FifoStorage::open ( int iType )
{
	char	szDir[ FS_MAX_PATH ];
	memset ( szDir, 0, sizeof(szDir) );
	//snprintf( szDir, sizeof(szDir) - 1, "%s/%05d/%s", m_strDataDir.c_str(), iType, getCompactDateString() );
	if (true == m_bDateSubdir) {	
		snprintf( szDir, sizeof(szDir) - 1, "%s/%05d/%s", m_strDataDir.c_str(), iType, date("Ymd") );
	}
	else {
		snprintf( szDir, sizeof(szDir) - 1, "%s/%05d", m_strDataDir.c_str(), iType );
	}
	  
	char 	szPath[ FS_MAX_PATH ];
	memset ( szPath, 0, sizeof(szPath) );
	//snprintf( szPath, sizeof(szPath) - 1, "%s/%s", szDir, getCompactDateTimeString() );
	snprintf( szPath, sizeof(szPath) - 1, "%s/%05d_%s%03d", szDir, iType, date("YmdHMS"), getMinorSerial() );
	
	// mkdir data directory
	int	iReturn;
	iReturn = mkpdir ( szDir );
	if (iReturn < 0)
	{
		ERROR ( CAT_SINKER, 0, "Fail to mkdir '%s' : %s", szDir, strerror(errno) );
		return	-1;
	}
	
	// open data file
	//DEBUG ( CAT_SINKER, "open() FifoStorage file '%s' ...", szPath );
	FILE*	_file = fopen ( szPath, "a" );
	if (NULL == _file)
	{
		ERROR ( CAT_SINKER, 0, "Fail to open file '%s' : %s.", szPath, strerror(errno) ); 
		return	-1;
	}
	fseek ( _file, 0, SEEK_END );
	setbuf ( _file, NULL );
	
	//DEBUG ( CAT_SINKER, "open() FifoStorage '%s' File %p", szPath, _file );
	
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		pCell = new FifoStorageCell();
		m_mapStorageCell[ iType ] = pCell;
	}
	else {
		pCell = result->second;
	}

	strncpy( pCell->m_szFileName, szPath, FS_MAX_PATH - 1 );
	pCell->m_file = _file;
	pCell->m_iSwitchCounter = 0;
	pCell->m_iSwitchTimeout = time(NULL) + m_iSwitchTimeout;
	pCell->m_iLineCount = 0;
	
	//DEBUG ( CAT_SINKER, "open(iType = %d) FifoStorage='%s' fp=%p", iType, pCell->m_szFileName, pCell->m_file );
	
	return	0;
}

int
FifoStorage::close ( int iType )
{
	
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		return	0;
	}
	else {
		pCell = result->second;
	}

	if (NULL != pCell->m_file)
	{
		//DEBUG ( CAT_SINKER, "close(iType = %d) FifoStorage='%s' fp=%p", iType, pCell->m_szFileName, pCell->m_file );

		fclose ( pCell->m_file );
		pCell->m_file = NULL;
		
		bool	bRenamed = true;		
		char	szNewFile[ FS_MAX_PATH ];
		memset( szNewFile, 0, sizeof(szNewFile) );
        int     iReturn;

        if ( true == m_bGZipFile ) {
            snprintf( szNewFile, sizeof(szNewFile) - 1, "%s.%s.gz", pCell->m_szFileName, m_szSwitchSuffix );
            iReturn = gz_compress( pCell->m_szFileName, szNewFile );		
        }
        else {
            snprintf( szNewFile, sizeof(szNewFile) - 1, "%s.%s", pCell->m_szFileName, m_szSwitchSuffix );
            iReturn = rename( pCell->m_szFileName, szNewFile );		
        }
		if ( iReturn < 0 ) {
			ERROR ( CAT_SINKER, 0, "Fail to rename file from '%s' to  '%s' : [ret=%d] %s.", pCell->m_szFileName, szNewFile, iReturn, strerror(errno) );
			bRenamed = false; 
		}
        else if ( true == m_bGZipFile ) {
            // unlink old file
            iReturn = unlink( pCell->m_szFileName );
            if ( iReturn < 0 ) {
                ERROR ( CAT_SINKER, 0, "Fail to unlink file '%s' : %s.", pCell->m_szFileName, strerror(errno) );
            }
        }
		
		FILE*	file = fopen ( m_strFifoList.c_str(), "a" );
		if (NULL != file)
		{
			if ( true == bRenamed ) {
				fprintf ( file, "%s\n", szNewFile );
				INFO ( CAT_STAT_SINK, "fifo_storage %s %d", szNewFile, pCell->m_iLineCount );
			}
			else {
				fprintf ( file, "%s\n", pCell->m_szFileName );
				INFO ( CAT_STAT_SINK, "fifo_storage %s %d", pCell->m_szFileName, pCell->m_iLineCount );
			}
			fclose ( file );
		}
	
		//DEBUG ( CAT_SINKER, "close(iType = %d) FifoStorage='%s' fp=%p", iType, pCell->m_szFileName, pCell->m_file );
	}
	
	return	0;
}

inline	FILE*
FifoStorage::getFileHandle ( int iType )
{
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		pCell = new FifoStorageCell();
		pCell->m_iSwitchTimeout = time(NULL) + m_iSwitchTimeout;
		m_mapStorageCell[ iType ] = pCell;
	}
	else {
		pCell = result->second;
	}
	
	if (NULL == pCell->m_file)
	{
	
		//DEBUG ( CAT_SINKER, "FifoStorage '%s' File %p", m_strDataFile.c_str(), m_file );

		this->open ( iType );
	}
	
	//DEBUG ( CAT_SINKER, "getFileHandle(iType = %d) FifoStorage='%s' fp=%p", iType, pCell->m_szFileName, pCell->m_file );
	
	return	pCell->m_file;
}

FifoStorageCell*
FifoStorage::getStorageCell ( int iType )
{
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		pCell = new FifoStorageCell ();
		pCell->m_iSwitchTimeout = time(NULL) + m_iSwitchTimeout;
		m_mapStorageCell[ iType ] = pCell;
	}
	else {
		pCell = result->second;
	}
	
	return	pCell;
}

int
FifoStorage::write ( unsigned iType, const void* pvData, unsigned iDataSize )
{
	FifoStorageCell*	pCell = getStorageCell( iType );
	if ( NULL != pCell ) {
		char	_szType[ 32 ];
		int		_iTypeLength = snprintf( _szType, sizeof(_szType) - 1, "%u", iType );
		*(_szType + _iTypeLength) = '\0';

		cache_item_t*	pItem = NULL;
		if ( true == m_bCIDPrefix ) {
			pItem = allocateCache( iDataSize + _iTypeLength + 2 );	//%u\t<data>\n
		}
		else {
			pItem = allocateCache( iDataSize + 1 );	//<data>\n
		} 
		if ( NULL != pItem ) {
			char*	_pszBuf = pItem->m_szData;
			int		_iLen = 0;
			if ( true == m_bCIDPrefix ) {
				_iLen = sprintf( _pszBuf, "%s\t", _szType );
			}
			_pszBuf += _iLen; 
			memmove( _pszBuf, pvData, iDataSize );
			_pszBuf += iDataSize;
			*_pszBuf = '\n';
			++_pszBuf;
			*_pszBuf = '\0';
			
			pCell->m_cache.push_back( pItem );
			
			//DEBUG ( CAT_SINKER, "write(iType=%d) cache %p : data-size:%d item-size:%d data:[\n1234567890123456789012345678901234567890\n%s]", iType, pItem, pItem->m_iDataSize, pItem->m_iCacheItemSize, pItem->m_szData );
			
			return	0;
		}
	}
	
	FILE*	file = this->getFileHandle( iType );
	if (NULL == file)
	{
		ERROR ( CAT_SINKER, 0, "Fail to get file handle for QUEUE storage type '%d'.", iType );
		return	-2;
	}

	/// todo : to lock file exclusively
		
	fseek ( file, 0, SEEK_END );
	
	// write hash ID
	if ( true == m_bCIDPrefix ) {
		fprintf ( file, "%u\t", iType );
	}
	
	// write data
	int		iReturn = fwrite ( pvData, iDataSize, 1, file );
	if (iReturn < 1)
	{
		ERROR ( CAT_SINKER, 0, "Fail to store data into QUEUE '%d' (data size:%u)", iType, iDataSize );
		fwrite ( "\n", 1, 1, file );
	}
	else 
	{
		fwrite ( "\n", 1, 1, file );
		this->countAndSwitchStorage( iType );
	}
	pCell->m_iLineCount += 1;
	
	/// todo : to unlock file
	
	return	0;	
}

bool
FifoStorage::needSwitchByCount ( int iType )
{
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		return	false;
	}
	else {
		pCell = result->second;
	}

	return	this->needSwitchByCount( pCell );
}

bool
FifoStorage::needSwitchByCount ( FifoStorageCell* pCell )
{
	if ((m_iSwitchCountLimit > 0) && (pCell->m_iSwitchCounter >= m_iSwitchCountLimit))
	{
		return	true;
	}
	
	return	false;
}

bool
FifoStorage::needSwitchBySize ( int iType )
{
	if (getSwitchSizeLimit() < 1)
	{
		// NO size limit
		return	false;
	}
	
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		return	false;
	}
	else {
		pCell = result->second;
	}

	return	this->needSwitchBySize( pCell );
}

bool
FifoStorage::needSwitchBySize ( FifoStorageCell* pCell )
{
	//INFO ( CAT_SINKER, "%s(CELL::filename=%s fp=%p) ...", __func__, pCell->m_szFileName, pCell->m_file );
	if (NULL != pCell->m_file)
	{
		long	lOffset = ftell ( pCell->m_file );
		//INFO ( CAT_SINKER, "    offset=%ld (fifo.switchsize = %ld)", lOffset, m_iSwitchSliceSize );
		if (lOffset >= m_iSwitchSliceSize) {
			return	true;
		}
	}
	
	return	false;
}

bool
FifoStorage::needSwitchByTimeout ( FifoStorageCell* pCell )
{
	if (NULL != pCell->m_file)
	{
		time_t	now = time(NULL);
		if (now > pCell->m_iSwitchTimeout) {
			return	true;
		}
	}
	
	return	false;
}

int
FifoStorage::countAndSwitchStorage ( int iType )
{
	FifoStorageCell*	pCell = NULL;

	map_storage_cell_t::iterator	result = m_mapStorageCell.find( iType );
	if ( m_mapStorageCell.end() == result ) {
		return	0;
	}
	else {
		pCell = result->second;
	}

	pCell->m_iSwitchCounter += 1;
	
	return	switchStorage( iType, pCell );
}

int
FifoStorage::switchStorage ( int iType, FifoStorageCell* pCell )
{
	if ( 
		(true == this->needSwitchBySize( pCell )) || 
		(true == this->needSwitchByCount( pCell )) ||
		(true == this->needSwitchByTimeout( pCell ))
		)
	{
		if (NULL != pCell->m_file)
		{
			fseek ( pCell->m_file, 0, SEEK_END );
			long	lOffset = ftell ( pCell->m_file );
			if (lOffset > 1024)
			{
		
				//DEBUG ( CAT_SINKER, "FifoStorage '%s' File %p", m_strDataFile.c_str(), m_file );
				
				this->close ( iType );

				//// 2007-12-20 : lazy-open pCell			
				////this->open ( iType );
		
				//DEBUG ( CAT_SINKER, "FifoStorage '%s' File %p", m_strDataFile.c_str(), m_file );
			}
		}
		
		pCell->m_iSwitchCounter = 0;
	}
	
	return	0;
}

int
FifoStorage::switchStorage ( bool bForce )
{
	// flush all data from CACHE to DISK 
	bool	bReturn = this->flushCache();
	if ( false == bReturn ) {
		ERROR ( CAT_SINKER, 0, "Fail to flush cache." );
		return	-1;
	}
		
	if ((true == bForce) )
	{
		// switch file name 
		map_storage_cell_t::iterator	it = m_mapStorageCell.begin();
		while( m_mapStorageCell.end() != it ) {
			int		iType = it->first;
			FifoStorageCell*	pCell = it->second;
			if (NULL != pCell->m_file)
			{
				fseek ( pCell->m_file, 0, SEEK_END );
				long	lOffset = ftell ( pCell->m_file );
				if (lOffset > 0)
				{
			
					DEBUG ( CAT_SINKER, "#%d FifoStorage '%s' File %p", iType, pCell->m_szFileName, pCell->m_file );
					
					this->close ( it->first );

					//// 2007-12-20 : lazy-open pCell			
					//// this->open ( it->first );
			
					DEBUG ( CAT_SINKER, "#%d FifoStorage '%s' File %p", iType, pCell->m_szFileName, pCell->m_file );
				}
			}
			
			pCell->m_iSwitchCounter = 0;
			++it;
		}	// while
	}
	
	return	0;
}

////////////////////////////////////////////////////////////////////////

bool
FifoStorage::flushCache ( void )
{
	int		iFlushItemsCount = 0;
	int		iFlushBytesSum = 0;
	int		iFlushDataSizeSum = 0;
	
	INFO ( CAT_SINKER, "out-cache-size:%d out-cache-used:%d out-cache-free:%d", m_iOutCacheSize, m_iOutCacheSizeUsed, m_iOutCacheSizeFree );

	map_storage_cell_t::iterator	it = m_mapStorageCell.begin();
	for( ;m_mapStorageCell.end() != it; ++it ) {
		int		iType = it->first;
		FifoStorageCell*	pCell = it->second;
		cache_item_list_t&	cachedItems = pCell->m_cache;
		
		if ( cachedItems.size() < 1 ) {
			switchStorage( iType, pCell );

			continue;
		}
		
		int		iBufSizeUsed = 0;
		int		iBufSizeLeft = m_iFlushBufSize;
		char*	pszBuf = m_pszFlushBuf;
		
		FILE*	file = this->getFileHandle( iType );
		if (NULL == file)
		{
			ERROR ( CAT_SINKER, 0, "Fail to get file handle for QUEUE storage type '%d'.", iType );
			return	false;
		}

		/// todo : to lock file exclusively
			
		fseek ( file, 0, SEEK_END );
		
		//DEBUG ( CAT_SINKER, "FifoStorage '%s' File %p", m_strDataFile.c_str(), m_file );
		for ( unsigned i=0; i<cachedItems.size(); i++ ) {
			cache_item_t*	pItem = cachedItems[ i ];
			
			++iFlushItemsCount;
			iFlushBytesSum += pItem->m_iCacheItemSize;
			iFlushDataSizeSum += pItem->m_iDataSize;
			pCell->m_iLineCount += 1;
			
			if ( iBufSizeLeft < pItem->m_iDataSize ) {
				// write data
				//DEBUG ( CAT_SINKER, "flush buffer ( type=%d size=%d ) :\n%s", iType, iBufSizeUsed, m_pszFlushBuf );
				
				int		iReturn = fwrite ( m_pszFlushBuf, iBufSizeUsed, 1, file );
				if (iReturn < 1)
				{
					ERROR ( CAT_SINKER, 0, "Fail to store data into QUEUE '%d' (data size:%u)", iType, iBufSizeUsed );
					return	false;
				}
				
				iBufSizeUsed = 0;
				iBufSizeLeft = m_iFlushBufSize;
				pszBuf = m_pszFlushBuf;
			}
			
			if ( iBufSizeLeft < pItem->m_iDataSize ) {
				// write some HUGE item directly 
				
				// write hash ID
				fprintf ( file, "%u\t", iType );
				
				// write data
				int		iReturn = fwrite ( pItem->m_szData, pItem->m_iDataSize, 1, file );
				if (iReturn < 1)
				{
					ERROR ( CAT_SINKER, 0, "Fail to store data DIRECTLY into QUEUE '%d' (data size:%u)", iType, pItem->m_iDataSize );
					fwrite ( "\n", 1, 1, file );
					return	false;
				}
				
				continue;
			}
			
			//DEBUG ( CAT_SINKER, "flushing cache-item (type=%d) %p : data-size=%d item-size=%d data:[%s]", iType, pItem, pItem->m_iDataSize, pItem->m_iCacheItemSize, pItem->m_szData );
			memmove( pszBuf, pItem->m_szData, pItem->m_iDataSize );
			pszBuf += pItem->m_iDataSize;
			iBufSizeUsed += pItem->m_iDataSize;
			iBufSizeLeft -= pItem->m_iDataSize;
			
			pCell->m_iSwitchCounter += 1;			
		}
		if ( iBufSizeUsed > 0 ) {
			// write data
			//DEBUG ( CAT_SINKER, "flush buffer ( type=%d size=%d ) :\n%s", iType, iBufSizeUsed, m_pszFlushBuf );

			int		iReturn = fwrite ( m_pszFlushBuf, iBufSizeUsed, 1, file );
			if (iReturn < 1)
			{
				ERROR ( CAT_SINKER, 0, "Fail to store data into QUEUE '%d' (data size:%u)", iType, iBufSizeUsed );
				return	false;
			}
			
			iBufSizeUsed = 0;
			iBufSizeLeft = m_iFlushBufSize;
			pszBuf = m_pszFlushBuf;
		}
		
		/// todo : to unlock file

		switchStorage( iType, pCell );

		// clear cache		
		pCell->m_cache.clear();
	}	// while
	
	m_iOutCacheSizeFree = m_iOutCacheSize;
	m_iOutCacheSizeUsed = 0;

	INFO ( CAT_SINKER, "flush-items:%d flush-bytes:%d cache-bytes:%d", iFlushItemsCount, iFlushDataSizeSum, iFlushBytesSum );
	
	return	true;
}

cache_item_t *
FifoStorage::allocateCache( int iDataSize )
{
	if ( iDataSize < 1 ) {
		return	NULL;
	}

	DEBUG ( CAT_SINKER, "out-cache-size:%d out-cache-used:%d out-cache-free:%d", m_iOutCacheSize, m_iOutCacheSizeUsed, m_iOutCacheSizeFree );

	if ( NULL == m_pszOutCache ) {
		char*	pszOutCache = (char *)malloc( m_iOutCacheSize * sizeof(char) );
		if ( NULL == pszOutCache ) {
			ERROR ( CAT_SINKER, 0, "Fail to allocate out cache (size=%d).", m_iOutCacheSize );
			return	NULL;
		}
		
		m_pszOutCache = pszOutCache;
		m_iOutCacheSizeFree = m_iOutCacheSize;
		m_iOutCacheSizeUsed = 0;
	}
		
	int	iCacheItemSize = iDataSize  & 0x7FFFFFFC;
	int	iReminder = iDataSize & 0x00000003;
	if ( iReminder > 0 ) {
		iCacheItemSize += 4;
	}
	iCacheItemSize += sizeof( cache_item_head_t );
	
	if ( m_iOutCacheSizeFree <= iCacheItemSize ) {
		// cache-buffer should have 1+ more free bytes 
		bool	bReturn = this->flushCache();
		if ( false == bReturn ) {
			ERROR ( CAT_SINKER, 0, "Fail to flush cache." );
			return	NULL;
		}
		if ( m_iOutCacheSizeFree <= iCacheItemSize ) {
			// cache-buffer is too small to hold data 
			return	NULL;
		}
	}
	
	void*	_pv = static_cast< void * >( m_pszOutCache + m_iOutCacheSizeUsed );
	cache_item_t*	pItem = static_cast< cache_item_t* >( _pv );
	m_iOutCacheSizeUsed += iCacheItemSize;
	m_iOutCacheSizeFree -= iCacheItemSize;
	
	pItem->m_iDataSize = iDataSize;
	pItem->m_iCacheItemSize = iCacheItemSize;
	
	return	pItem;
}

////////////////////////////////////////////////////////////////////////

static char*
strtolower( char* s, int len )
{
	char	*b, *e;
	
	b = s;
	e = s + len;
	
	while( b < e ) {
		*b = tolower( *b );
		b++;
	}
	
	return	s;
}

static	int
trim ( char* pszString )
{
    char*   p = pszString;
    if ( NULL == p ) {
        return  0;
    }

    char*   q = pszString;
    char*   e = p + strlen(p) - 1;
    while ((' ' == *p) || ('\t' == *p) || ('\r' == *p) || ('\n' == *p)) p++;
    while ((p < e) && ((' ' == *e) || ('\t' == *e) || ('\r' == *e) || ('\n' == *e))) e--;

    int     len = e - p + 1;
    while (p <= e) *q++ = *p++;
    *q = '\0';

    return  len;
}


int
FifoStorage::loadConfig ( const char* pszConfigFile )
{
	if (NULL == pszConfigFile)
	{
		ERROR ( CAT_SINKER, 0, "No configure file appointed." );

		INFO ( CAT_SINKER, "fifo.switchsize = %ld", m_iSwitchSliceSize );
		INFO ( CAT_SINKER, "fifo.switchcount = %d", m_iSwitchCountLimit );
		INFO ( CAT_SINKER, "fifo.switchtimeout = %d", m_iSwitchTimeout );
		INFO ( CAT_SINKER, "fifo.outcachesize = %d", m_iOutCacheSize );
		if ( m_bCIDPrefix ) {
			INFO ( CAT_SINKER, "fifo.cidprefix = true" );
		}
		else {
			INFO ( CAT_SINKER, "fifo.cidprefix = false" );
		}
		if ( m_bDateSubdir ) {
			INFO ( CAT_SINKER, "fifo.datesubdir = true" );
		}
		else {
			INFO ( CAT_SINKER, "fifo.datesubdir = false" );
		}
		INFO ( CAT_SINKER, "fifo.switchsuffix = '%s'", m_szSwitchSuffix );

		return	0;
	}
	
	FILE*   file = fopen ( pszConfigFile, "r" );
	if (NULL == file)
	{
		ERROR ( CAT_SINKER, 0, "Fail to open configure file '%s' : [%d] %s.", pszConfigFile, errno, strerror(errno) );

		INFO ( CAT_SINKER, "fifo.switchsize = %ld", m_iSwitchSliceSize );
		INFO ( CAT_SINKER, "fifo.switchcount = %d", m_iSwitchCountLimit );
		INFO ( CAT_SINKER, "fifo.switchtimeout = %d", m_iSwitchTimeout );
		INFO ( CAT_SINKER, "fifo.outcachesize = %d", m_iOutCacheSize );
		if ( m_bCIDPrefix ) {
			INFO ( CAT_SINKER, "fifo.cidprefix = true" );
		}
		else {
			INFO ( CAT_SINKER, "fifo.cidprefix = false" );
		}
		if ( m_bDateSubdir ) {
			INFO ( CAT_SINKER, "fifo.datesubdir = true" );
		}
		else {
			INFO ( CAT_SINKER, "fifo.datesubdir = false" );
		}
        if ( m_bGZipFile ) {
            INFO ( CAT_SINKER, "fifo.gzipfile = true" );
        }
        else {
            INFO ( CAT_SINKER, "fifo.gzipfile = false" );
        }
		INFO ( CAT_SINKER, "fifo.switchsuffix = '%s'", m_szSwitchSuffix );

		return	-1;
	}
	
	while ( !feof(file) ) {
		char    szLine[ 1024 ];
		memset ( szLine, 0, sizeof(szLine) );
		fgets ( szLine, sizeof(szLine), file );
		if ( (trim (szLine) < 1) || ('#' == szLine[0]) )
			continue;

		char    *pszKey = szLine, *pszValue = szLine, *p = szLine;

		while ( ('=' != *p) && ('\0' != *p) ) p++;
		if ( '\0' == *p )
			continue;
		*p = '\0';
		pszValue = ++p;

		while ( ('#' != *p) && ('\0' != *p) ) p++;
		if ( '#' == *p )
			*p = '\0';

		trim ( pszKey ); trim ( pszValue );
		//fprintf ( stdout, "key='%s' value='%s'\n", pszKey, pszValue );
		
        if (0 == strcmp("sinkroot", pszKey))
        {
            if (NULL != m_pszSinkRoot)
            {
                free ( m_pszSinkRoot );
                m_pszSinkRoot = NULL;
            }
    
            m_pszSinkRoot = strdup ( pszValue );
        }
    

		if (0 == strcmp("fifo.switchsize", pszKey))
		{
			long	_iSwitchSliceSize = atol ( pszValue );
			if ( _iSwitchSliceSize < 1 ) {
				ERROR ( CAT_SINKER, 0, "WRONG CONFIG %s = %s.", pszKey, pszValue );
			}
			else {
				m_iSwitchSliceSize = _iSwitchSliceSize;
			}
			
			continue;
		}
		
		if (0 == strcmp("fifo.switchcount", pszKey))
		{
			int		_iSwitchCount = atoi ( pszValue );
			if ( _iSwitchCount < 0 ) {
				ERROR ( CAT_SINKER, 0, "WRONG CONFIG %s = %s.", pszKey, pszValue );
			}
			else {
				m_iSwitchCountLimit = _iSwitchCount;
			}
			
			continue;
		}
		
		if (0 == strcmp("fifo.switchtimeout", pszKey))
		{
			int		_iSwitchTimeout = atoi ( pszValue );
			if ( _iSwitchTimeout < 0 ) {
				ERROR ( CAT_SINKER, 0, "WRONG CONFIG %s = %s.", pszKey, pszValue );
			}
			else {
				m_iSwitchTimeout = _iSwitchTimeout;
			}
			
			continue;
		}
		
		if (0 == strcmp("fifo.switchsuffix", pszKey))
		{
			if ( '\0' != *pszValue ) {
				int		_iLen = snprintf( m_szSwitchSuffix, sizeof(m_szSwitchSuffix) - 1, "%s", pszValue );
				m_szSwitchSuffix[ _iLen ] = '\0';
			}
				
			continue;
		}
		
		if (0 == strcmp("fifo.outcachesize", pszKey))
		{
			int		_iCacheSize = atoi ( pszValue );
			if ( _iCacheSize < 0 ) {
				ERROR ( CAT_SINKER, 0, "WRONG CONFIG %s = %s.", pszKey, pszValue );
			}
			else {
				m_iOutCacheSize = _iCacheSize;
				m_iOutCacheSizeFree = m_iOutCacheSize;
			}
			
			continue;
		}
		
		if ( 0 == strcmp("fifo.cidprefix", pszKey) ) 
		{
			strtolower( pszValue, strlen(pszValue) );
			if ( (0 == strcmp("y", pszValue)) || (0 == strcmp("yes", pszValue))
				|| (0 == strcmp("t", pszValue)) || (0 == strcmp("true", pszValue)) 
				) {
				m_bCIDPrefix = true;
			}
			else {
				m_bCIDPrefix = false;
			}
		}	// fifo.cidprefix
		
		if ( 0 == strcmp("fifo.datesubdir", pszKey) ) 
		{
			strtolower( pszValue, strlen(pszValue) );
			if ( (0 == strcmp("y", pszValue)) || (0 == strcmp("yes", pszValue))
				|| (0 == strcmp("t", pszValue)) || (0 == strcmp("true", pszValue)) 
				) {
				m_bDateSubdir = true;
			}
			else {
				m_bDateSubdir = false;
			}
		}	// fifo.datesubdir
		
		if ( 0 == strcmp("fifo.gzipfile", pszKey) ) 
		{
			strtolower( pszValue, strlen(pszValue) );
			if ( (0 == strcmp("y", pszValue)) || (0 == strcmp("yes", pszValue))
				|| (0 == strcmp("t", pszValue)) || (0 == strcmp("true", pszValue)) 
				) {
				m_bGZipFile = true;
			}
			else {
				m_bGZipFile = false;
			}
		}	// fifo.gzipfile
		
	}

	fclose ( file );
	
	m_pszConfigFile = pszConfigFile;

	INFO ( CAT_SINKER, "fifo.switchsize = %ld", m_iSwitchSliceSize );
	INFO ( CAT_SINKER, "fifo.switchcount = %d", m_iSwitchCountLimit );
	INFO ( CAT_SINKER, "fifo.switchtimeout = %d", m_iSwitchTimeout );
	INFO ( CAT_SINKER, "fifo.outcachesize = %d", m_iOutCacheSize );
	if ( m_bCIDPrefix ) {
		INFO ( CAT_SINKER, "fifo.cidprefix = true" );
	}
	else {
		INFO ( CAT_SINKER, "fifo.cidprefix = false" );
	}
	if ( m_bDateSubdir ) {
		INFO ( CAT_SINKER, "fifo.datesubdir = true" );
	}
	else {
		INFO ( CAT_SINKER, "fifo.datesubdir = false" );
	}
	if ( m_bGZipFile ) {
		INFO ( CAT_SINKER, "fifo.gzipfile = true" );
	}
	else {
		INFO ( CAT_SINKER, "fifo.gzipfile = false" );
	}
	INFO ( CAT_SINKER, "fifo.switchsuffix = '%s'", m_szSwitchSuffix );

	return  0;
}


